package streamAPI.source;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.util.concurrent.TimeUnit;

/**
 * 自定义支持多并行度 的 Source
 *
 * @author yue.cao
 * @since 09-28-2020
 */
@Slf4j
public class MyRichParallelSource extends RichParallelSourceFunction<Long> {

	private Long count = 1L;
	private boolean isRunning = true;


	/** 此方法只会运行一次 但是是在每一个线程中运行一次 */
	@Override
	public void open(Configuration parameters) throws Exception {
		super.open(parameters);
		log.warn("====Cy open====");
	}

	@Override
	public void close() throws Exception {
		super.close();
		log.warn("====Cy close====");
	}

	@Override
	public void run(SourceContext<Long> ctx) throws Exception {
		while (isRunning) {
			ctx.collect(count++);
			TimeUnit.SECONDS.sleep(1);
		}
	}

	@Override
	public void cancel() {
		isRunning = false;
		log.warn("====Cy cancel====");
	}
}
